package cn.com.rocketmq.broadCast;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;

import java.util.List;

/**
 * 广播模式
 * @author xuhaixiang
 * @date 2023-09-10
 */
public class Consumer {

    /**
     * 日志对象
     */
    private static  final Logger logger = LoggerFactory.getLogger(Consumer.class);


    public static void main(String[] args) {
        DefaultMQPushConsumer consumer = null;
        try {
            consumer = new DefaultMQPushConsumer("group_rocketmq_asyncDemo");
            consumer.setNamesrvAddr("192.168.113.101:9876");
            //参数一：topic、参数二：过滤（*表示不过滤）
            consumer.subscribe("rocketmq_asyncDemo", "*");
            //设置为广播模式
            consumer.setMessageModel(MessageModel.BROADCASTING);
            //设置消息监听
            //MessageListenerConcurrently  并发消费
            //MessageListenerOrderly 顺序消费
            consumer.setMessageListener(new MessageListenerConcurrently() {
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                    list.forEach(item -> {
                        try {
                            logger.info("消息消费成功！消息ID={}，消息内容：{}", item.getMsgId(), new String(item.getBody(), "UTF-8"));
                        } catch (Exception e) {
                            logger.error("消息消费失败！", e);
                        }
                    });
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            //消费者启动
            consumer.start();
        } catch (MQClientException e) {
            logger.error("消费者消费异常！",e);
        }
    }
}
